package com.shujia.spark.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.junit.{Before, Test}

case class Student(id: String, name: String, age: Int, gender: String, clazz: String)

case class Score(id: String, subjectId: String, sco: Int)

case class Subject(subjectId: String, subjectName: String, fullSco: Int)

class Demo20StuExercise extends Serializable {


  var student: RDD[Student] = _
  var score: RDD[Score] = _
  var subject: RDD[Subject] = _

  var stuNameAndClazzMap: collection.Map[String, String] = _
  var subjectFullScoMap: collection.Map[String, Int] = _

  @Before
  def init(): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo20StuExercise")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    student = sc.textFile("data/stu/students.txt")
      .map(line => {
        val splits: Array[String] = line.split(",")
        Student(splits(0), splits(1), splits(2).toInt, splits(3), splits(4))
      })

    score = sc.textFile("data/stu/score.txt")
      .map(line => {
        val splits: Array[String] = line.split(",")
        Score(splits(0), splits(1), splits(2).toInt)
      })

    subject = sc.textFile("data/stu/subject.txt")
      .map(line => {
        val splits: Array[String] = line.split(",")
        Subject(splits(0), splits(1), splits(2).toInt)
      })

    stuNameAndClazzMap = student.map(stu => (stu.id, s"${stu.name},${stu.clazz}")).collectAsMap()

    // 以科目id作为key，科目总分作为value形成Map
    subjectFullScoMap = subject.map(s => (s.subjectId, s.fullSco)).collectAsMap()
  }

  def filterJoinAndPrintWithIds(ids: Array[String]): Unit = {
    // 先构建两个Map
    // 1、以学生id作为Key，姓名+班级作为Value stuNameAndClazzMap
    // 2、以科目id作为Key，科目名作为Value subjectNameMap
    //    val stuNameAndClazzMap: Map[String, String] = student.map(stu => (stu.id, s"${stu.name},${stu.clazz}")).toMap
    val subjectNameMap: collection.Map[String, String] = subject.map(sub => (sub.subjectId, sub.subjectName)).collectAsMap()

    // 遍历分数
    score
      // 将符合条件的学生分数过滤出来
      .filter(sco => ids.contains(sco.id))
      // 借助两个Map实现关联
      .map(sco => {
        val id: String = sco.id
        val subjectId: String = sco.subjectId
        val nameAndClazz: String = stuNameAndClazzMap.getOrElse(id, "")
        val subjectName: String = subjectNameMap.getOrElse(subjectId, "")
        s"$id,$nameAndClazz,$subjectName,${sco.sco}"
      })
      .foreach(println)
  }

  @Test
  // 1、统计年级排名前十学生各科的分数 [学号,  姓名，班级，科目，分数]
  def question01(): Unit = {
    val idArr: Array[String] = score
      .map(s => (s.id, s.sco))
      .reduceByKey(_ + _)
      .sortBy(_._2, ascending = false)
      .map(_._1)
      .take(10)

    filterJoinAndPrintWithIds(idArr)
  }

  @Test
  // 2、统计总分大于年级平均分的学生 [学号，姓名，班级，总分]
  def question2(): Unit = {

    val avgSco: Double = score.map(_.sco).reduce(_ + _) / student.count().toDouble

    println(avgSco) // 年级平均分

    score
      .map(s => (s.id, s.sco))
      .reduceByKey(_ + _)
      .filter(kv => kv._2 > avgSco)
      .map(kv => {
        val id: String = kv._1
        val nameAndClazz: String = stuNameAndClazzMap.getOrElse(id, "")
        s"$id,$nameAndClazz,${kv._2}"
      })
      .foreach(println)

  }

  @Test
  // 3、统计每科都及格的学生 [学号，姓名，班级，科目，分数]
  def question3(): Unit = {

    // 遍历分数数据
    val allPassIds: Array[String] = score
      // 将不及格的分数过滤掉，及格的保留
      .filter(sco => {
        val fullSco: Int = subjectFullScoMap.getOrElse(sco.subjectId, 1000000000)
        sco.sco.toDouble / fullSco >= 0.6
      })
      // 统计每个学生及格的科目数量
      .map(sco => (sco.id, 1))
      .reduceByKey(_ + _)
      // 过滤出科目都及格的学生
      .filter(_._2 == 6)
      .map(_._1)
      .collect()

    filterJoinAndPrintWithIds(allPassIds)

  }

  @Test
  // 统计每个班级的前三名 [学号，姓名，班级，分数]
  def question04(): Unit = {
    score
      .map(sco => (sco.id, sco.sco))
      .reduceByKey(_ + _)
      .map(kv => {
        val nameAndClazz: String = stuNameAndClazzMap.getOrElse(kv._1, "")
        val splits: Array[String] = nameAndClazz.split(",")
        (kv._1, splits(0), splits(1), kv._2)
      })
      .groupBy(_._3)
      .flatMap(kv => {
        kv._2
          .toList
          .sortBy(-_._4)
          .take(3)
      }).foreach(println)
  }

  @Test
  // 5、统计偏科最严重的前100名学生  [学号，姓名，班级，科目，分数]
  def question05(): Unit = {
    val idList: Array[String] = score
      .groupBy(_.id)
      .map(kv => {
        // 将每个分数转换成百分制
        val newScoIter: Iterable[Double] = kv
          ._2
          .map(s => {
            s.sco * 100.toDouble / subjectFullScoMap.getOrElse(s.subjectId, 100)
          })

        val avgSco: Double = newScoIter.sum / newScoIter.size

        (kv._1, newScoIter.map(newSco => Math.pow(newSco - avgSco, 2)).sum)
      })
      .sortBy(_._2, ascending = false)
      .map(_._1)
      .take(100)

    filterJoinAndPrintWithIds(idList)
  }
}
